package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.beans.TableProcess;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.util.*;

/**
 * @author Felix
 * @date 2023/12/18
 * 过滤出维度数据
 */
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, TableProcess, JSONObject> {

    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    private Map<String,TableProcess> configMap = new HashMap<>();

    public TableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //将配置表中的配置信息预加载到程序中
        //注册驱动
        Class.forName("com.mysql.cj.jdbc.Driver");
        //建立连接
        Connection conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/gmall0710_config?" +
            "user=root&password=123456&useUnicode=true&characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");
        //获取数据库操作对象
        String sql = "select * from gmall0710_config.table_process_dim ";
        PreparedStatement ps = conn.prepareStatement(sql);
        //执行SQL语句
        ResultSet rs = ps.executeQuery();
        ResultSetMetaData metaData = rs.getMetaData();
        //处理结果集
        while (rs.next()){
            //定义一个json对象，用于封装遍历出来的每一条配置信息
            JSONObject jsonObj = new JSONObject();
            //遍历所有的列，将列名作为jsonObj的key
            for (int i = 1; i <= metaData.getColumnCount(); i++) {
                String columnName = metaData.getColumnName(i);
                Object columnValue = rs.getObject(i);
                jsonObj.put(columnName,columnValue);
            }
            TableProcess tableProcess = JSON.toJavaObject(jsonObj, TableProcess.class);
            //将读取到的一行配置信息  放到configMap中
            configMap.put(tableProcess.getSourceTable(),tableProcess);
        }

        //释放资源
        rs.close();
        ps.close();
        conn.close();
    }

    //处理主流业务数据
    @Override
    public void processElement(JSONObject jsonObj, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        // {"database":"gmall0710","xid":1628,"data":{"tm_name":"Redmi","create_time":"2021-12-14 00:00:00","logo_url":"aaa","id":1},"old":{},"commit":true,"type":"update","table":"base_trademark","ts":1702867046}
        // System.out.println(jsonObj);
        //获取处理的业务数据库表的表名
        String table = jsonObj.getString("table");
        //获取广播状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //根据表名到广播状态中获取对应的配置信息
        TableProcess tableProcess = null;


        if((tableProcess = broadcastState.get(table)) != null
            ||(tableProcess = configMap.get(table)) != null){
            //如果从广播状态中找到了对应的配置信息，说明是维度数据

            //将维度数据 data部分内容传递到下游
            // {"tm_name":"Redmi","create_time":"2021-12-14 00:00:00","logo_url":"aaabbb","id":1}
            JSONObject dataJsonObj = jsonObj.getJSONObject("data");

            //在向下游传递数据前，过滤掉不需要传递的字段
            // {"tm_name":"Redmi","id":1}
            String sinkColumns = tableProcess.getSinkColumns();
            filterColumns(dataJsonObj,sinkColumns);

            //在向下游传递数据前 补充操作类型
            String type = jsonObj.getString("type");
            dataJsonObj.put("type", type);

            //在向下游传递数据前 补充当前数据对应的配置信息
            dataJsonObj.put("table_process",tableProcess);

            // {"tm_name":"Redmi","id":1,"type":"update","table_process":{"op":"r","sinkColumns":"id,tm_name","sinkFamily":"info","sinkRowKey":"id","sinkTable":"dim_base_trademark","sourceTable":"base_trademark"}}
            out.collect(dataJsonObj);
        }

    }

    //过滤掉不需要向下游传递的属性
    // dataJsonObj :{"tm_name":"Redmi","create_time":"2021-12-14 00:00:00","logo_url":"aaabbb","id":1}
    // sinkColumns :id,tm_name
    private void filterColumns(JSONObject dataJsonObj, String sinkColumns) {
        String[] columnArr = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columnArr);
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        /*Iterator<Map.Entry<String, Object>> it = entrySet.iterator();

        for (;it.hasNext();) {
            Map.Entry<String, Object> entry = it.next();
            if(!columnList.contains(entry.getKey())){
                it.remove();
            }
        }*/

        entrySet.removeIf(entry->!columnList.contains(entry.getKey()));

    }

    //处理广播流配置数据
    @Override
    public void processBroadcastElement(TableProcess tableProcess, Context ctx, Collector<JSONObject> out) throws Exception {
        //获取对配置表进行操作的类型
        String op = tableProcess.getOp();
        //获取广播状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        String key = tableProcess.getSourceTable();
        if("d".equals(op)){
            //说明从配置表中删除了一条配置信息  将这条对应的配置从广播状态中删除掉
            broadcastState.remove(key);
            configMap.remove(key);

        }else{
            //说明对配置表进行了删除外的其它操作(r/c/u)  将最新的配置放到广播状态中去
            broadcastState.put(key,tableProcess);
            configMap.put(key,tableProcess);
        }
    }
}
